package channel

import (
	"context"
	"errors"
	"gitee.com/lipore/go-nio/pkg/events"
	"gitee.com/lipore/go-nio/pkg/transport"
	"gitee.com/lipore/plume/logger"
	"github.com/panjf2000/ants/v2"
	"sync"
	"time"
)

type (
	Group interface {
		ID() int64
		Add(c Channel) error
		Remove(c Channel) error
		WaitMessage()
		Close() error
	}

	group struct {
		id           int64
		ctx          context.Context
		cancel       context.CancelFunc
		channels     sync.Map
		pipeline     *Pipeline
		tsPool       transport.TransportPool
		runPool      *ants.PoolWithFunc
		readyChannel chan int64
	}
)

func NewGroup(ctx context.Context, pl *Pipeline, maxGoroutines int, id int64) (Group, error) {
	ctxWith, cancel := context.WithCancel(ctx)
	g := &group{
		id:           id,
		ctx:          ctxWith,
		cancel:       cancel,
		tsPool:       transport.NewTransportPool(),
		readyChannel: make(chan int64),
		pipeline:     pl,
	}
	p, err := ants.NewPoolWithFunc(maxGoroutines, g.readMessage)
	if err != nil {
		return nil, err
	}
	g.runPool = p
	go g.readWaiter()
	go g.WaitMessage()
	return g, nil
}

func (g *group) ID() int64 {
	return g.id
}

func (g *group) Pipeline(p *Pipeline) {
	g.pipeline = p
}

func (g *group) Add(c Channel) error {
	g.channels.Store(c.ID(), c)
	c.SetPipeline(g.pipeline)
	c.Pipeline().FireChannelEvent(c, events.ActiveEvent)
	return g.tsPool.Add(c.Transport())
}

func (g *group) Remove(c Channel) error {
	c.Pipeline().FireChannelEvent(c, events.InactiveEvent)
	if _, loaded := g.channels.LoadAndDelete(c.ID()); loaded {
		//if p := c2d.(Channel).Pipeline(); p != nil {
		//	p.FireChannelEvent(c, events.InactiveEvent)
		//}
		return g.tsPool.Remove(c.Transport())
	}
	return nil
}

func (g *group) WaitMessage() {
	for {
		select {
		case <-g.ctx.Done():
			return
		default:
			ts, closedTs, err := g.tsPool.WaitMessage()
			if err != nil {
				logger.Warnf("WaitMessage: %v", err)
				continue
			}
			for _, closedT := range closedTs {
				if c, exist := g.channels.LoadAndDelete(closedT.ID()); exist {
					g.pipeline.FireChannelEvent(c.(Channel), events.DisconnectEvent)
				}
			}
			for _, t := range ts {
				g.readyChannel <- t.ID()
			}
		}
	}
}

func (g *group) readWaiter() {
	for {
		select {
		case <-g.ctx.Done():
			return
		default:
			chId := <-g.readyChannel
			for {
				err := g.runPool.Invoke(chId)
				if err != nil {
					if errors.Is(ants.ErrPoolClosed, err) {
						return
					}
					if errors.Is(ants.ErrPoolOverload, err) {
						time.Sleep(1 * time.Millisecond)
						continue
					}
				}
				break
			}
		}
	}
}

func (g *group) Close() error {
	g.cancel()
	g.runPool.Release()
	return nil
}

func (g *group) readMessage(i interface{}) {
	chId := i.(int64)
	if c, ok := g.channels.Load(chId); ok {
		ch := c.(Channel)
		ch.Lock()
		rawMessage := make([]byte, 1024)
		for {
			if n, err := ch.Transport().Read(rawMessage); err == nil && n > 0 {
				_, err = ch.InboundBuffer().Write(rawMessage[:n])
				if err != nil {
					logger.Warnf("%v", err)
					_ = ch.Close()
				}
			} else {
				break
			}
		}
		ch.Pipeline().FireChannelInbound(ch)
		ch.Unlock()
	}
}

type Groups []Group

func (s Groups) MoveChannelTo(c Channel, id int64) error {
	for _, g := range s {
		var err error
		if id != g.ID() {
			err = g.Remove(c)
		} else {
			err = g.Add(c)
		}
		if err != nil {
			logger.Warnf("%v", err)
			return err
		}
	}
	return nil

}
